#include "binlog_manager.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <time.h>
#include <sys/stat.h>
#include <zlib.h>
// #include <uuid/uuid.h> // Optional for demo

// CRC32 calculation using zlib
uint32_t binlog_calculate_crc32(const void* data, uint32_t size) {
    return crc32(0, (const Bytef*)data, size);
}

// Generate binlog filename with sequence number
char* binlog_generate_filename(const char* prefix, uint32_t sequence) {
    char* filename = malloc(256);
    if (!filename) return NULL;
    
    snprintf(filename, 256, "%s.%06u", prefix, sequence);
    return filename;
}

// Parse sequence number from binlog filename
int binlog_parse_filename(const char* filename, uint32_t* sequence) {
    const char* dot = strrchr(filename, '.');
    if (!dot) return -1;
    
    return sscanf(dot + 1, "%u", sequence) == 1 ? 0 : -1;
}

// Event type to string conversion
const char* binlog_event_type_to_string(binlog_event_type_t type) {
    switch (type) {
        case UNKNOWN_EVENT: return "UNKNOWN_EVENT";
        case START_EVENT_V3: return "START_EVENT_V3";
        case QUERY_EVENT: return "QUERY_EVENT";
        case STOP_EVENT: return "STOP_EVENT";
        case ROTATE_EVENT: return "ROTATE_EVENT";
        case INTVAR_EVENT: return "INTVAR_EVENT";
        case XID_EVENT: return "XID_EVENT";
        case FORMAT_DESCRIPTION_EVENT: return "FORMAT_DESCRIPTION_EVENT";
        case TABLE_MAP_EVENT: return "TABLE_MAP_EVENT";
        case WRITE_ROWS_EVENT: return "WRITE_ROWS_EVENT";
        case UPDATE_ROWS_EVENT: return "UPDATE_ROWS_EVENT";
        case DELETE_ROWS_EVENT: return "DELETE_ROWS_EVENT";
        case GTID_LOG_EVENT: return "GTID_LOG_EVENT";
        case ANONYMOUS_GTID_LOG_EVENT: return "ANONYMOUS_GTID_LOG_EVENT";
        case PREVIOUS_GTIDS_LOG_EVENT: return "PREVIOUS_GTIDS_LOG_EVENT";
        default: return "UNKNOWN";
    }
}

// GTID Set Management
gtid_set_t* gtid_set_create(void) {
    gtid_set_t* set = malloc(sizeof(gtid_set_t));
    if (!set) return NULL;
    
    set->gtids = NULL;
    set->count = 0;
    set->capacity = 0;
    
    if (pthread_mutex_init(&set->mutex, NULL) != 0) {
        free(set);
        return NULL;
    }
    
    return set;
}

void gtid_set_destroy(gtid_set_t* set) {
    if (!set) return;
    
    pthread_mutex_destroy(&set->mutex);
    if (set->gtids) {
        free(set->gtids);
    }
    free(set);
}

int gtid_set_add(gtid_set_t* set, const gtid_t* gtid) {
    if (!set || !gtid) return -1;
    
    pthread_mutex_lock(&set->mutex);
    
    // Check if GTID already exists
    for (uint32_t i = 0; i < set->count; i++) {
        if (strcmp(set->gtids[i].server_uuid, gtid->server_uuid) == 0 &&
            set->gtids[i].transaction_id == gtid->transaction_id) {
            pthread_mutex_unlock(&set->mutex);
            return 0; // Already exists
        }
    }
    
    // Expand capacity if needed
    if (set->count >= set->capacity) {
        uint32_t new_capacity = set->capacity ? set->capacity * 2 : 16;
        gtid_t* new_gtids = malloc(new_capacity * sizeof(gtid_t));
        if (!new_gtids) {
            pthread_mutex_unlock(&set->mutex);
            return -1;
        }
        
        if (set->gtids) {
            memcpy(new_gtids, set->gtids, set->count * sizeof(gtid_t));
            free(set->gtids);
        }
        
        set->gtids = new_gtids;
        set->capacity = new_capacity;
    }
    
    // Add new GTID
    memcpy(&set->gtids[set->count], gtid, sizeof(gtid_t));
    set->count++;
    
    pthread_mutex_unlock(&set->mutex);
    return 0;
}

bool gtid_set_contains(gtid_set_t* set, const gtid_t* gtid) {
    if (!set || !gtid) return false;
    
    pthread_mutex_lock(&set->mutex);
    
    for (uint32_t i = 0; i < set->count; i++) {
        if (strcmp(set->gtids[i].server_uuid, gtid->server_uuid) == 0 &&
            set->gtids[i].transaction_id == gtid->transaction_id) {
            pthread_mutex_unlock(&set->mutex);
            return true;
        }
    }
    
    pthread_mutex_unlock(&set->mutex);
    return false;
}

char* gtid_set_to_string(gtid_set_t* set) {
    if (!set) return NULL;
    
    pthread_mutex_lock(&set->mutex);
    
    if (set->count == 0) {
        pthread_mutex_unlock(&set->mutex);
        char* empty = malloc(1);
        if (empty) empty[0] = '\0';
        return empty;
    }
    
    // Calculate required buffer size
    size_t buffer_size = 0;
    for (uint32_t i = 0; i < set->count; i++) {
        buffer_size += strlen(set->gtids[i].server_uuid) + 32; // UUID + ":1-" + number + ","
    }
    
    char* result = malloc(buffer_size);
    if (!result) {
        pthread_mutex_unlock(&set->mutex);
        return NULL;
    }
    
    result[0] = '\0';
    for (uint32_t i = 0; i < set->count; i++) {
        if (i > 0) strcat(result, ",");
        char gtid_str[128];
        snprintf(gtid_str, sizeof(gtid_str), "%s:1-%llu", 
                set->gtids[i].server_uuid, set->gtids[i].transaction_id);
        strcat(result, gtid_str);
    }
    
    pthread_mutex_unlock(&set->mutex);
    return result;
}

gtid_set_t* gtid_set_from_string(const char* gtid_string) {
    if (!gtid_string) return NULL;
    
    gtid_set_t* set = gtid_set_create();
    if (!set) return NULL;
    
    // Parse GTID string format: uuid:1-N,uuid:1-M,...
    char* str_copy = strdup(gtid_string);
    if (!str_copy) {
        gtid_set_destroy(set);
        return NULL;
    }
    
    char* token = strtok(str_copy, ",");
    while (token) {
        char* colon = strchr(token, ':');
        if (colon) {
            *colon = '\0';
            char* range = colon + 1;
            
            // Parse range (simplified - assume 1-N format)
            uint64_t max_id;
            if (sscanf(range, "1-%llu", &max_id) == 1) {
                gtid_t gtid;
                strncpy(gtid.server_uuid, token, sizeof(gtid.server_uuid) - 1);
                gtid.server_uuid[sizeof(gtid.server_uuid) - 1] = '\0';
                gtid.transaction_id = max_id;
                
                gtid_set_add(set, &gtid);
            }
        }
        token = strtok(NULL, ",");
    }
    
    free(str_copy);
    return set;
}

// Event Management
binlog_event_t* binlog_event_create(binlog_event_type_t type, void* data, uint32_t data_size) {
    binlog_event_t* event = malloc(sizeof(binlog_event_t));
    if (!event) return NULL;
    
    memset(event, 0, sizeof(binlog_event_t));
    
    // Initialize header
    event->header.timestamp = time(NULL);
    event->header.type_code = type;
    event->header.server_id = 1; // Will be set by manager
    event->header.event_length = BINLOG_HEADER_SIZE + data_size + BINLOG_CHECKSUM_SIZE;
    event->header.next_position = 0; // Will be set when writing
    event->header.flags = 0;
    
    // Copy data if provided
    if (data && data_size > 0) {
        event->data = malloc(data_size);
        if (!event->data) {
            free(event);
            return NULL;
        }
        memcpy(event->data, data, data_size);
        event->owns_data = true;
    } else {
        event->data = data;
        event->owns_data = false;
    }
    
    event->data_size = data_size;
    
    // Calculate checksum
    event->checksum = binlog_event_calculate_checksum(event);
    
    return event;
}

void binlog_event_destroy(binlog_event_t* event) {
    if (!event) return;
    
    if (event->owns_data && event->data) {
        free(event->data);
    }
    free(event);
}

binlog_event_t* binlog_event_copy(binlog_event_t* event) {
    if (!event) return NULL;
    
    return binlog_event_create(event->header.type_code, event->data, event->data_size);
}

uint32_t binlog_event_calculate_checksum(binlog_event_t* event) {
    if (!event) return 0;
    
    // Calculate checksum over header + data
    uint32_t header_crc = binlog_calculate_crc32(&event->header, BINLOG_HEADER_SIZE);
    if (event->data && event->data_size > 0) {
        return binlog_calculate_crc32(event->data, event->data_size) ^ header_crc;
    }
    return header_crc;
}

// File operations
static int create_new_binlog_file(binlog_manager_t* binlog) {
    if (!binlog) return -1;
    
    // Generate filename
    char* filename = binlog_generate_filename(binlog->filename_prefix, binlog->next_sequence);
    if (!filename) return -1;
    
    char full_path[512];
    snprintf(full_path, sizeof(full_path), "%s/%s", binlog->binlog_directory, filename);
    
    // Create file
    int fd = open(full_path, O_CREAT | O_WRONLY | O_TRUNC, 0644);
    if (fd < 0) {
        free(filename);
        return -1;
    }
    
    // Write binlog magic number
    if (write(fd, BINLOG_MAGIC, 4) != 4) {
        close(fd);
        free(filename);
        return -1;
    }
    
    // Create binlog_file structure
    binlog_file_t* file = malloc(sizeof(binlog_file_t));
    if (!file) {
        close(fd);
        free(filename);
        return -1;
    }
    
    file->filename = filename;
    file->fd = fd;
    file->size = 4; // Magic number size
    file->position = 4;
    file->sequence = binlog->next_sequence;
    file->is_active = true;
    file->needs_sync = false;
    file->created_time = time(NULL);
    file->next = NULL;
    
    if (pthread_mutex_init(&file->mutex, NULL) != 0) {
        close(fd);
        free(filename);
        free(file);
        return -1;
    }
    
    // Link to file list
    pthread_rwlock_wrlock(&binlog->files_lock);
    
    if (binlog->active_file) {
        binlog->active_file->is_active = false;
    }
    
    file->next = binlog->files;
    binlog->files = file;
    binlog->active_file = file;
    binlog->next_sequence++;
    binlog->files_created++;
    
    pthread_rwlock_unlock(&binlog->files_lock);
    
    return 0;
}

static int write_event_to_file(binlog_file_t* file, binlog_event_t* event) {
    if (!file || !event) return -1;
    
    pthread_mutex_lock(&file->mutex);
    
    // Update header with current position
    event->header.next_position = file->position + event->header.event_length;
    
    // Write header
    ssize_t written = write(file->fd, &event->header, BINLOG_HEADER_SIZE);
    if (written != BINLOG_HEADER_SIZE) {
        pthread_mutex_unlock(&file->mutex);
        return -1;
    }
    
    // Write data if present
    if (event->data && event->data_size > 0) {
        written = write(file->fd, event->data, event->data_size);
        if (written != event->data_size) {
            pthread_mutex_unlock(&file->mutex);
            return -1;
        }
    }
    
    // Write checksum
    written = write(file->fd, &event->checksum, BINLOG_CHECKSUM_SIZE);
    if (written != BINLOG_CHECKSUM_SIZE) {
        pthread_mutex_unlock(&file->mutex);
        return -1;
    }
    
    // Update file position and size
    file->position = event->header.next_position;
    file->size = file->position;
    file->needs_sync = true;
    
    pthread_mutex_unlock(&file->mutex);
    return 0;
}

static binlog_event_t* read_event_from_file(binlog_file_t* file, uint64_t position) {
    if (!file) return NULL;
    
    pthread_mutex_lock(&file->mutex);
    
    // Seek to position
    if (lseek(file->fd, position, SEEK_SET) != position) {
        pthread_mutex_unlock(&file->mutex);
        return NULL;
    }
    
    // Read header
    binlog_event_header_t header;
    if (read(file->fd, &header, BINLOG_HEADER_SIZE) != BINLOG_HEADER_SIZE) {
        pthread_mutex_unlock(&file->mutex);
        return NULL;
    }
    
    // Validate event length
    if (header.event_length < BINLOG_HEADER_SIZE + BINLOG_CHECKSUM_SIZE ||
        header.event_length > BINLOG_MAX_EVENT_SIZE) {
        pthread_mutex_unlock(&file->mutex);
        return NULL;
    }
    
    // Calculate data size
    uint32_t data_size = header.event_length - BINLOG_HEADER_SIZE - BINLOG_CHECKSUM_SIZE;
    
    // Read data if present
    void* data = NULL;
    if (data_size > 0) {
        data = malloc(data_size);
        if (!data) {
            pthread_mutex_unlock(&file->mutex);
            return NULL;
        }
        
        if (read(file->fd, data, data_size) != data_size) {
            free(data);
            pthread_mutex_unlock(&file->mutex);
            return NULL;
        }
    }
    
    // Read checksum
    uint32_t checksum;
    if (read(file->fd, &checksum, BINLOG_CHECKSUM_SIZE) != BINLOG_CHECKSUM_SIZE) {
        if (data) free(data);
        pthread_mutex_unlock(&file->mutex);
        return NULL;
    }
    
    pthread_mutex_unlock(&file->mutex);
    
    // Create event
    binlog_event_t* event = malloc(sizeof(binlog_event_t));
    if (!event) {
        if (data) free(data);
        return NULL;
    }
    
    event->header = header;
    event->data = data;
    event->data_size = data_size;
    event->checksum = checksum;
    event->owns_data = (data != NULL);
    event->next = NULL;
    
    // Verify checksum
    uint32_t calculated_checksum = binlog_event_calculate_checksum(event);
    if (calculated_checksum != checksum) {
        binlog_event_destroy(event);
        return NULL;
    }
    
    return event;
}

static uint32_t assign_transaction_id(binlog_manager_t* binlog) {
    if (!binlog) return 0;
    
    pthread_mutex_lock(&binlog->manager_mutex);
    uint64_t id = binlog->next_transaction_id++;
    pthread_mutex_unlock(&binlog->manager_mutex);
    
    return id;
}

// Background maintenance thread
void* binlog_maintenance_thread_func(void* arg) {
    binlog_manager_t* binlog = (binlog_manager_t*)arg;
    if (!binlog) return NULL;
    
    while (binlog->maintenance_running) {
        pthread_mutex_lock(&binlog->maintenance_mutex);
        
        // Wait for work or timeout
        struct timespec timeout;
        clock_gettime(CLOCK_REALTIME, &timeout);
        timeout.tv_sec += 60; // Wake up every minute
        
        pthread_cond_timedwait(&binlog->maintenance_cond, &binlog->maintenance_mutex, &timeout);
        
        if (!binlog->maintenance_running) {
            pthread_mutex_unlock(&binlog->maintenance_mutex);
            break;
        }
        
        pthread_mutex_unlock(&binlog->maintenance_mutex);
        
        // Sync dirty files
        pthread_rwlock_rdlock(&binlog->files_lock);
        binlog_file_t* file = binlog->files;
        while (file) {
            pthread_mutex_lock(&file->mutex);
            if (file->needs_sync) {
                fsync(file->fd);
                file->needs_sync = false;
                binlog->sync_count++;
            }
            pthread_mutex_unlock(&file->mutex);
            file = file->next;
        }
        pthread_rwlock_unlock(&binlog->files_lock);
        
        // Check for old files to purge
        if (binlog->expire_logs_days > 0) {
            time_t cutoff_time = time(NULL) - (binlog->expire_logs_days * 24 * 60 * 60);
            binlog_purge_old_files(binlog, cutoff_time);
        }
    }
    
    return NULL;
}

// Main Binlog Manager API Implementation

binlog_manager_t* binlog_manager_create(const char* binlog_dir, const char* prefix, uint32_t server_id) {
    if (!binlog_dir || !prefix) return NULL;
    
    binlog_manager_t* binlog = malloc(sizeof(binlog_manager_t));
    if (!binlog) return NULL;
    
    memset(binlog, 0, sizeof(binlog_manager_t));
    
    // Initialize configuration
    binlog->binlog_directory = strdup(binlog_dir);
    binlog->filename_prefix = strdup(prefix);
    binlog->server_id = server_id;
    binlog->max_file_size = BINLOG_FILE_SIZE;
    binlog->sync_binlog = true;
    binlog->expire_logs_days = 7; // Default 7 days
    binlog->next_sequence = 1;
    binlog->next_transaction_id = 1;
    
    // Generate server UUID (simplified for demo)
    snprintf(binlog->server_uuid, sizeof(binlog->server_uuid), 
             "550e8400-e29b-41d4-a716-%012u", server_id);
    
    // Initialize GTID set
    binlog->executed_gtids = gtid_set_create();
    if (!binlog->executed_gtids) {
        free(binlog->binlog_directory);
        free(binlog->filename_prefix);
        free(binlog);
        return NULL;
    }
    
    // Initialize synchronization
    if (pthread_mutex_init(&binlog->manager_mutex, NULL) != 0 ||
        pthread_rwlock_init(&binlog->files_lock, NULL) != 0 ||
        pthread_mutex_init(&binlog->maintenance_mutex, NULL) != 0 ||
        pthread_cond_init(&binlog->maintenance_cond, NULL) != 0 ||
        pthread_mutex_init(&binlog->cache_mutex, NULL) != 0) {
        
        gtid_set_destroy(binlog->executed_gtids);
        free(binlog->binlog_directory);
        free(binlog->filename_prefix);
        free(binlog);
        return NULL;
    }
    
    // Initialize event cache
    binlog->cache_size = 1000; // Default cache size
    binlog->event_cache = malloc(binlog->cache_size * sizeof(binlog_event_t*));
    if (!binlog->event_cache) {
        pthread_mutex_destroy(&binlog->manager_mutex);
        pthread_rwlock_destroy(&binlog->files_lock);
        pthread_mutex_destroy(&binlog->maintenance_mutex);
        pthread_cond_destroy(&binlog->maintenance_cond);
        pthread_mutex_destroy(&binlog->cache_mutex);
        gtid_set_destroy(binlog->executed_gtids);
        free(binlog->binlog_directory);
        free(binlog->filename_prefix);
        free(binlog);
        return NULL;
    }
    
    // Create binlog directory if it doesn't exist
    struct stat st;
    if (stat(binlog_dir, &st) != 0) {
        if (mkdir(binlog_dir, 0755) != 0) {
            binlog_manager_destroy(binlog);
            return NULL;
        }
    }
    
    return binlog;
}

void binlog_manager_destroy(binlog_manager_t* binlog) {
    if (!binlog) return;
    
    // Stop maintenance thread
    if (binlog->maintenance_running) {
        binlog_manager_stop(binlog);
    }
    
    // Clean up files
    pthread_rwlock_wrlock(&binlog->files_lock);
    binlog_file_t* file = binlog->files;
    while (file) {
        binlog_file_t* next = file->next;
        
        pthread_mutex_destroy(&file->mutex);
        if (file->fd >= 0) {
            close(file->fd);
        }
        free(file->filename);
        free(file);
        
        file = next;
    }
    pthread_rwlock_unlock(&binlog->files_lock);
    
    // Clean up event cache
    pthread_mutex_destroy(&binlog->cache_mutex);
    if (binlog->event_cache) {
        for (uint32_t i = 0; i < binlog->cache_used; i++) {
            if (binlog->event_cache[i]) {
                binlog_event_destroy(binlog->event_cache[i]);
            }
        }
        free(binlog->event_cache);
    }
    
    // Clean up synchronization
    pthread_mutex_destroy(&binlog->manager_mutex);
    pthread_rwlock_destroy(&binlog->files_lock);
    pthread_mutex_destroy(&binlog->maintenance_mutex);
    pthread_cond_destroy(&binlog->maintenance_cond);
    
    // Clean up GTID set
    gtid_set_destroy(binlog->executed_gtids);
    
    // Clean up strings
    free(binlog->binlog_directory);
    free(binlog->filename_prefix);
    
    free(binlog);
}

int binlog_manager_start(binlog_manager_t* binlog) {
    if (!binlog) return -1;
    
    // Create initial binlog file
    if (create_new_binlog_file(binlog) != 0) {
        return -1;
    }
    
    // Write format description event
    if (binlog_write_format_desc_event(binlog) != 0) {
        return -1;
    }
    
    // Start maintenance thread
    binlog->maintenance_running = true;
    if (pthread_create(&binlog->maintenance_thread, NULL, binlog_maintenance_thread_func, binlog) != 0) {
        binlog->maintenance_running = false;
        return -1;
    }
    
    return 0;
}

int binlog_manager_stop(binlog_manager_t* binlog) {
    if (!binlog) return -1;
    
    // Signal maintenance thread to stop
    pthread_mutex_lock(&binlog->maintenance_mutex);
    binlog->maintenance_running = false;
    pthread_cond_signal(&binlog->maintenance_cond);
    pthread_mutex_unlock(&binlog->maintenance_mutex);
    
    // Wait for maintenance thread
    pthread_join(binlog->maintenance_thread, NULL);
    
    // Sync all files
    pthread_rwlock_rdlock(&binlog->files_lock);
    binlog_file_t* file = binlog->files;
    while (file) {
        pthread_mutex_lock(&file->mutex);
        if (file->fd >= 0) {
            fsync(file->fd);
        }
        pthread_mutex_unlock(&file->mutex);
        file = file->next;
    }
    pthread_rwlock_unlock(&binlog->files_lock);
    
    return 0;
}

// Configuration functions
int binlog_set_max_file_size(binlog_manager_t* binlog, uint64_t size) {
    if (!binlog || size < 1024) return -1;
    
    pthread_mutex_lock(&binlog->manager_mutex);
    binlog->max_file_size = size;
    pthread_mutex_unlock(&binlog->manager_mutex);
    
    return 0;
}

int binlog_set_server_id(binlog_manager_t* binlog, uint32_t server_id) {
    if (!binlog) return -1;
    
    pthread_mutex_lock(&binlog->manager_mutex);
    binlog->server_id = server_id;
    pthread_mutex_unlock(&binlog->manager_mutex);
    
    return 0;
}

int binlog_set_server_uuid(binlog_manager_t* binlog, const char* uuid) {
    if (!binlog || !uuid) return -1;
    
    pthread_mutex_lock(&binlog->manager_mutex);
    strncpy(binlog->server_uuid, uuid, sizeof(binlog->server_uuid) - 1);
    binlog->server_uuid[sizeof(binlog->server_uuid) - 1] = '\0';
    pthread_mutex_unlock(&binlog->manager_mutex);
    
    return 0;
}

int binlog_set_sync_mode(binlog_manager_t* binlog, bool sync_enabled) {
    if (!binlog) return -1;
    
    pthread_mutex_lock(&binlog->manager_mutex);
    binlog->sync_binlog = sync_enabled;
    pthread_mutex_unlock(&binlog->manager_mutex);
    
    return 0;
}

int binlog_set_expire_days(binlog_manager_t* binlog, uint32_t days) {
    if (!binlog) return -1;
    
    pthread_mutex_lock(&binlog->manager_mutex);
    binlog->expire_logs_days = days;
    pthread_mutex_unlock(&binlog->manager_mutex);
    
    return 0;
}

// Event logging functions
int binlog_write_query_event(binlog_manager_t* binlog, uint32_t thread_id,
                             const char* db_name, const char* query,
                             uint32_t exec_time, uint16_t error_code) {
    if (!binlog || !query) return -1;
    
    // Create query event data
    query_event_data_t query_data;
    query_data.thread_id = thread_id;
    query_data.exec_time = exec_time;
    query_data.error_code = error_code;
    query_data.status_vars_len = 0;
    
    if (db_name) {
        query_data.db_name_len = strlen(db_name);
        query_data.db_name = (char*)db_name;
    } else {
        query_data.db_name_len = 0;
        query_data.db_name = NULL;
    }
    
    query_data.query_len = strlen(query);
    query_data.query = (char*)query;
    
    // Calculate total data size
    uint32_t data_size = sizeof(query_event_data_t) + query_data.db_name_len + query_data.query_len;
    
    // Create event
    binlog_event_t* event = binlog_event_create(QUERY_EVENT, &query_data, data_size);
    if (!event) return -1;
    
    event->header.server_id = binlog->server_id;
    
    // Write to active file
    int result = write_event_to_file(binlog->active_file, event);
    if (result == 0) {
        binlog->events_written++;
        binlog->bytes_written += event->header.event_length;
        
        // Check if file rotation is needed
        if (binlog->active_file->size >= binlog->max_file_size) {
            binlog_rotate_file(binlog);
        }
    }
    
    binlog_event_destroy(event);
    return result;
}

int binlog_write_xid_event(binlog_manager_t* binlog, uint64_t xid) {
    if (!binlog) return -1;
    
    xid_event_data_t xid_data;
    xid_data.xid = xid;
    
    binlog_event_t* event = binlog_event_create(XID_EVENT, &xid_data, sizeof(xid_data));
    if (!event) return -1;
    
    event->header.server_id = binlog->server_id;
    
    int result = write_event_to_file(binlog->active_file, event);
    if (result == 0) {
        binlog->events_written++;
        binlog->bytes_written += event->header.event_length;
    }
    
    binlog_event_destroy(event);
    return result;
}

int binlog_write_gtid_event(binlog_manager_t* binlog, const gtid_t* gtid) {
    if (!binlog || !gtid) return -1;
    
    gtid_event_data_t gtid_data;
    gtid_data.flags = 0;
    memcpy(gtid_data.server_uuid, gtid->server_uuid, 16); // Binary format
    gtid_data.transaction_id = gtid->transaction_id;
    gtid_data.logical_clock_type = 2; // LOGICAL_CLOCK_TYPE_LOGICAL
    gtid_data.logical_clock = 0;
    
    binlog_event_t* event = binlog_event_create(GTID_LOG_EVENT, &gtid_data, sizeof(gtid_data));
    if (!event) return -1;
    
    event->header.server_id = binlog->server_id;
    
    int result = write_event_to_file(binlog->active_file, event);
    if (result == 0) {
        binlog->events_written++;
        binlog->bytes_written += event->header.event_length;
        
        // Add to executed GTIDs
        gtid_set_add(binlog->executed_gtids, gtid);
    }
    
    binlog_event_destroy(event);
    return result;
}

int binlog_write_format_desc_event(binlog_manager_t* binlog) {
    if (!binlog) return -1;
    
    format_desc_event_data_t format_data;
    format_data.binlog_version = BINLOG_VERSION;
    strncpy(format_data.server_version, "GuDB-1.0.0", sizeof(format_data.server_version) - 1);
    format_data.server_version[sizeof(format_data.server_version) - 1] = '\0';
    format_data.create_timestamp = time(NULL);
    format_data.common_header_len = BINLOG_HEADER_SIZE;
    format_data.post_header_len = NULL; // Simplified
    format_data.checksum_alg = 1; // CRC32
    
    binlog_event_t* event = binlog_event_create(FORMAT_DESCRIPTION_EVENT, &format_data, sizeof(format_data));
    if (!event) return -1;
    
    event->header.server_id = binlog->server_id;
    
    int result = write_event_to_file(binlog->active_file, event);
    if (result == 0) {
        binlog->events_written++;
        binlog->bytes_written += event->header.event_length;
    }
    
    binlog_event_destroy(event);
    return result;
}

int binlog_rotate_file(binlog_manager_t* binlog) {
    if (!binlog) return -1;
    
    binlog->files_rotated++;
    return create_new_binlog_file(binlog);
}

int binlog_purge_old_files(binlog_manager_t* binlog, time_t before_time) {
    if (!binlog) return -1;
    
    pthread_rwlock_wrlock(&binlog->files_lock);
    
    binlog_file_t* prev = NULL;
    binlog_file_t* current = binlog->files;
    
    while (current) {
        if (current->created_time < before_time && !current->is_active) {
            // Remove from list
            if (prev) {
                prev->next = current->next;
            } else {
                binlog->files = current->next;
            }
            
            // Close and delete file
            pthread_mutex_destroy(&current->mutex);
            close(current->fd);
            
            char full_path[512];
            snprintf(full_path, sizeof(full_path), "%s/%s", binlog->binlog_directory, current->filename);
            unlink(full_path);
            
            free(current->filename);
            free(current);
            
            current = prev ? prev->next : binlog->files;
        } else {
            prev = current;
            current = current->next;
        }
    }
    
    pthread_rwlock_unlock(&binlog->files_lock);
    return 0;
}

// Statistics and monitoring
void binlog_manager_print_stats(binlog_manager_t* binlog) {
    if (!binlog) return;
    
    pthread_mutex_lock(&binlog->manager_mutex);
    
    printf("Binlog Manager Statistics:\n");
    printf("  Server ID: %u\n", binlog->server_id);
    printf("  Server UUID: %s\n", binlog->server_uuid);
    printf("  Events written: %llu\n", binlog->events_written);
    printf("  Bytes written: %llu\n", binlog->bytes_written);
    printf("  Files created: %llu\n", binlog->files_created);
    printf("  Files rotated: %llu\n", binlog->files_rotated);
    printf("  Sync operations: %llu\n", binlog->sync_count);
    printf("  Max file size: %llu bytes\n", binlog->max_file_size);
    printf("  Expire days: %u\n", binlog->expire_logs_days);
    printf("  Sync mode: %s\n", binlog->sync_binlog ? "enabled" : "disabled");
    
    pthread_mutex_unlock(&binlog->manager_mutex);
    
    // Print GTID set
    char* gtid_str = gtid_set_to_string(binlog->executed_gtids);
    if (gtid_str) {
        printf("  Executed GTIDs: %s\n", gtid_str);
        free(gtid_str);
    }
}

uint64_t binlog_get_total_size(binlog_manager_t* binlog) {
    if (!binlog) return 0;
    
    uint64_t total_size = 0;
    
    pthread_rwlock_rdlock(&binlog->files_lock);
    binlog_file_t* file = binlog->files;
    while (file) {
        pthread_mutex_lock(&file->mutex);
        total_size += file->size;
        pthread_mutex_unlock(&file->mutex);
        file = file->next;
    }
    pthread_rwlock_unlock(&binlog->files_lock);
    
    return total_size;
}

uint32_t binlog_get_file_count(binlog_manager_t* binlog) {
    if (!binlog) return 0;
    
    uint32_t count = 0;
    
    pthread_rwlock_rdlock(&binlog->files_lock);
    binlog_file_t* file = binlog->files;
    while (file) {
        count++;
        file = file->next;
    }
    pthread_rwlock_unlock(&binlog->files_lock);
    
    return count;
}

gtid_set_t* binlog_get_executed_gtids(binlog_manager_t* binlog) {
    if (!binlog) return NULL;
    return binlog->executed_gtids;
}

bool binlog_is_mysql_compatible(binlog_manager_t* binlog) {
    return binlog != NULL; // Our implementation is MySQL compatible
}